package com.thp.bigdata.webClick.mr;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.thp.bigdata.webClick.mrBean.PageViewsBean;
import com.thp.bigdata.webClick.mrBean.VisitBean;

/**
 * 从PageViews模型数据结果中进一步梳理出visit模型
 * 
 * 经过这里之后出去的数据格式：
 * sessionid   satrt-time   out-time   satrt-page   out-page   pagecounts  ...
 * 
 * @author 汤小萌
 *
 */
public class ClickStreamVisit {
	
	static class ClickStreamVisitMapper extends Mapper<LongWritable, Text, Text, PageViewsBean> {
		PageViewsBean pvBean = new PageViewsBean();
		Text k = new Text();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fields = line.split("\001");
			int step = Integer.parseInt(fields[5]);   // 访问的步数
			pvBean.set(fields[0], fields[1], fields[2], fields[3],fields[4], step, fields[6], fields[7], fields[8], fields[9]);
			k.set(pvBean.getSession());
			context.write(k, pvBean);
		}
	}
	
	
	static class ClickStreamVisitReducer extends Reducer<Text, PageViewsBean, NullWritable, VisitBean> {
		@Override
		protected void reduce(Text session, Iterable<PageViewsBean> pvBeans, Context context)
				throws IOException, InterruptedException {
			// 将pvBean按照step排序
			ArrayList<PageViewsBean> pvBeanList = new ArrayList<PageViewsBean>();
			for(PageViewsBean pvBean : pvBeans) {
				PageViewsBean bean = new PageViewsBean();
				try {
					BeanUtils.copyProperties(bean, pvBean);
					pvBeanList.add(bean);
				} catch (IllegalAccessException | InvocationTargetException e) {
					e.printStackTrace();
				}
			}
			Collections.sort(pvBeanList, new Comparator<PageViewsBean>() {
				@Override
				public int compare(PageViewsBean o1, PageViewsBean o2) {
					return o1.getStep() > o2.getStep() ? 1 : -1;
				}
			});
			
			// 取这次visit的首尾pageViews记录，放入VisitBean中
			VisitBean visitBean = new VisitBean();
			// 取visit 的首记录
			visitBean.setInPage(pvBeanList.get(0).getRequest());
			visitBean.setInTime(pvBeanList.get(0).getTimeStr());
			// 取visit 的尾记录
			visitBean.setOutPage(pvBeanList.get(pvBeanList.size() - 1).getRequest());
			visitBean.setOutTime(pvBeanList.get(pvBeanList.size() - 1).getTimeStr());
			// visit访问的页面数
			visitBean.setPageVisits(pvBeanList.size());
			// 来访者的ip
			visitBean.setRemote_addr(pvBeanList.get(0).getRemote_addr());
			// 本次visit的referal
			visitBean.setReferal(pvBeanList.get(0).getReferal());
			visitBean.setSession(session.toString());
			
			context.write(NullWritable.get(), visitBean);
		}
	}
	
	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(ClickStreamVisit.class);

		job.setMapperClass(ClickStreamVisitMapper.class);
		job.setReducerClass(ClickStreamVisitReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(PageViewsBean.class);

		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(VisitBean.class);
		
		
//		FileInputFormat.setInputPaths(job, new Path(args[0]));
//		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		FileInputFormat.setInputPaths(job, new Path("f:/weblog_1/pageviews"));
		FileOutputFormat.setOutputPath(job, new Path("f:/weblog_1/visitout"));
		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("f:/weblog_1/visitout"))) {
			fs.delete(new Path("f:/weblog_1/visitout"), true);
		}
		
		boolean res = job.waitForCompletion(true);
		System.exit(res?0:1);
	}
	
	/**
	 * 	
	 * 2018年11月29日 上午9:00:57
	 * @param a
	 */
	public void a1(int a) {
		new StringBuffer();
	}
	
}
